package com.liuyjy.es.config;

import com.liuyjy.es.service.EsService;
import com.liuyjy.es.service.EsServiceImpl;
import com.liuyjy.es.utils.EsCacheManager;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.http.HttpHost;
import org.apache.http.HttpRequestInterceptor;
import org.apache.http.auth.AuthScope;
import org.apache.http.auth.UsernamePasswordCredentials;
import org.apache.http.client.CredentialsProvider;
import org.apache.http.impl.client.BasicCredentialsProvider;
import org.elasticsearch.action.DocWriteRequest;
import org.elasticsearch.action.bulk.*;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestClient;
import org.elasticsearch.client.RestClientBuilder;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.common.unit.ByteSizeUnit;
import org.elasticsearch.common.unit.ByteSizeValue;
import org.elasticsearch.core.TimeValue;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import java.util.Arrays;
import java.util.List;
import java.util.stream.Collectors;

/**
 * @Author liuyjy
 * @Date 2022/3/22 14:11
 * @Description: 配置文件
 **/
// EsSecurityConfig.java
@Configuration
//@EnableConfigurationProperties(ESProperties.class)
@Slf4j
public class ESConfig {

    @Bean
    public ESProperties esProperties() {
        return new ESProperties();
    }

    @Bean
    public EsService esService() {
        return new EsServiceImpl();
    }
    @Bean
    public EsCacheManager cacheManager(){
        return new EsCacheManager();
    }

    /**
     * 创建并配置RestHighLevelClient bean
     * 该bean用于与Elasticsearch进行交互，并在应用关闭时关闭连接
     * @return RestHighLevelClient实例，用于执行高级REST操作
     */
    @Bean(destroyMethod = "close")
    public RestHighLevelClient restHighLevelClient() {
        ESProperties properties = esProperties();
        // 根据配置的节点地址创建HttpHost数组
        HttpHost[] hosts = Arrays.stream(properties.getNodes().split(","))
                .map(node -> new HttpHost(node.split(":")[0], Integer.parseInt(node.split(":")[1])))
                .toArray(HttpHost[]::new);

        // 创建ResticBuilder并配置连接参数
        RestClientBuilder builder = RestClient.builder(hosts)
                .setRequestConfigCallback(requestConfigBuilder -> requestConfigBuilder
                        .setConnectTimeout(properties.getConnectionTimeout())
                        .setSocketTimeout(properties.getSocketTimeout()))
                .setHttpClientConfigCallback(httpClientBuilder -> {
                    // 根据配置的用户名和密码设置认证信息
                    if (StringUtils.isNotBlank(properties.getUsername())) {
                        CredentialsProvider credentialsProvider = new BasicCredentialsProvider();
                        credentialsProvider.setCredentials(AuthScope.ANY,
                                new UsernamePasswordCredentials(properties.getUsername(), properties.getPassword()));
                        httpClientBuilder.setDefaultCredentialsProvider(credentialsProvider);

                    } else if (StringUtils.isNotBlank(properties.getApiKey())) {
                        // 如果配置了API密钥，则设置认证头
                        httpClientBuilder.addInterceptorLast((HttpRequestInterceptor) (request, context) ->
                                request.addHeader("Authorization", "ApiKey " + properties.getApiKey()));
                    }
                    // 配置HTTP客户端的最大连接数和每路由最大连接数
//                    return httpClientBuilder.setMaxConnTotal(properties.getMaxConnTotal())
//                            .setMaxConnPerRoute(properties.getMaxConnPerRoute());
                    // 连接池参数
                    return httpClientBuilder
                            .setMaxConnTotal(properties.getMaxConnTotal())
                            .setMaxConnPerRoute(properties.getMaxConnPerRoute())
                            //.setRetryHandler(new RetryHandler(properties.getMaxRetries()))
                            ;
                });

        // 创建并返回RestHighLevelClient实例
        return new RestHighLevelClient(builder);
    }

    /**
     * 批量更新
     *
     * @return BulkProcessor
     */
    @Bean
    public BulkProcessor bulkProcessor() {

        BulkProcessor.Listener listener = new BulkProcessor.Listener() {
            @Override
            public void beforeBulk(long executionId, BulkRequest request) {
                log.info("1. 【beforeBulk】批次[{}] 携带 {} 请求数量", executionId, request.numberOfActions());
            }

            @Override
            public void afterBulk(long executionId, BulkRequest request,
                                  BulkResponse response) {
                if (!response.hasFailures()) {
                    log.info("2. 【afterBulk-成功】批量 [{}] 完成在 {} ms", executionId, response.getTook().getMillis());
                } else {
                    BulkItemResponse[] items = response.getItems();
                    for (BulkItemResponse item : items) {
                        if (item.isFailed()) {
                            log.info("2. 【afterBulk-失败】批量 [{}] 出现异常的原因 : {}", executionId, item.getFailureMessage());
                            break;
                        }
                    }
                }
            }

            @Override
            public void afterBulk(long executionId, BulkRequest request,
                                  Throwable failure) {

                List<DocWriteRequest<?>> requests = request.requests();
                List<String> esIds = requests.stream().map(DocWriteRequest::id).collect(Collectors.toList());
                log.error("3. 【afterBulk-failure失败】es执行bluk失败,失败的esId为：{}", esIds, failure);
            }
        };

        BulkProcessor.Builder builder = BulkProcessor.builder(((bulkRequest, bulkResponseActionListener) -> {
            restHighLevelClient().bulkAsync(bulkRequest, RequestOptions.DEFAULT, bulkResponseActionListener);
        }), listener);
        //到达10000条时刷新
        builder.setBulkActions(10000);
        //内存到达8M时刷新
        builder.setBulkSize(new ByteSizeValue(8L, ByteSizeUnit.MB));
        //设置的刷新间隔10s
        builder.setFlushInterval(TimeValue.timeValueSeconds(10));
        //设置允许执行的并发请求数。
        builder.setConcurrentRequests(8);
        //设置重试策略
        builder.setBackoffPolicy(BackoffPolicy.constantBackoff(TimeValue.timeValueSeconds(1), 3));
        return builder.build();
    }

    private HttpHost[] parseHosts(String nodes) {
        return Arrays.stream(nodes.split(","))
                .map(node -> new HttpHost(node.split(":")[0], Integer.parseInt(node.split(":")[1])))
                .toArray(HttpHost[]::new);
    }


    // 熔断器配置
//    @Bean
//    public CircuitBreaker esCircuitBreaker(ESProperties properties) {
//        return CircuitBreaker.of("es-circuit", CircuitBreakerConfig.custom()
//                .failureRateThreshold(properties.getCircuitBreakerThreshold())
//                .waitDurationInOpenState(Duration.ofSeconds(30))
//                .build());
//    }
}
